package com.powernode.demo;

import com.powernode.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.time.LocalDateTime;
import java.util.List;

/**
 * 发送事务消息
 * @author jiaxXM
 * @date 2024/7/9
 */
public class JTransactiotEest {

    @Test
    public void testTransactionProducer() throws Exception {

        // 创建一个事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-group");
        // 设置nameServer地址
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        //设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            // 这个是执行本地业务方法
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
                System.out.println(LocalDateTime.now());
                System.out.println(new String(msg.getBody()));
                // 这个可以使用try catch对业务代码进行性包裹
                // COMMIT_MESSAGE 表示允许消费者消费该消息
                // ROLLBACK_MESSAGE 表示该消息将被删除，不允许消费
                // UNKNOW表示需要MQ回查才能确定状态 那么过一会 代码会走下面的checkLocalTransaction(msg)方法
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 这里是回查方法 回查不是再次执行业务操作，而是确认上面的操作是否有结果
             *  默认是1min回查 默认回查15次 超过次数则丢弃打印日志 可以通过参数设置
             *  transactionTimeOut 超时时间
             *  transactionCheckMax 最大回查次数
             *  transactionCheckInterval 回查间隔时间单位毫秒
             *  触发条件
             *      1.当上面执行本地事务返回结果UNKNOW时,或者下面的回查方法也返回UNKNOW时 会触发回查
             *      2.当上面操作超过20s没有做出一个结果，也就是超时或者卡主了，也会进行回查
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.err.println(LocalDateTime.now());
                System.err.println(new String(msg.getBody()));
                // 这里
                return LocalTransactionState.UNKNOW;
            }
        });
        producer.start();
        Message message = new Message("transactionTopic", "我是一个事务消息".getBytes());
        // 发送消息
        producer.sendMessageInTransaction(message, null);
        System.out.println("发送消息====================================="+LocalDateTime.now());
        System.in.read();
    }

    @Test
    public void testTransactionConsumer() throws Exception {
        // 创建默认消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        // 设置nameServer地址
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // 订阅一个主题来消费   *表示没有过滤参数 表示这个主题的任何消息
        consumer.subscribe("transactionTopic", "*");
        // 注册一个消费监听 MessageListenerConcurrently是并发消费
        // 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                // 这里执行消费的代码 默认是多线程消费
                System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}
